package com.justgame.mq.server.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.justgame.mq.client.constant.RabbitConfigConstant;
import com.justgame.mq.client.service.RabbitMqSendService;
import com.justgame.project.cloud.base.entity.MgMessageQueueLog;
import com.rabbitmq.client.AMQP;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.UUID;

/**
 * @author ZhouWeiBo
 * @create 2022/5/30
 */
@Service
@Slf4j
public class RabbitMqSendServiceImpl implements RabbitMqSendService {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private ConnectionFactory connectionFactory;

    /**
     * RabbitMQ发送消息(使用默认的交换机)
     *
     * @param routingKey : 路由键
     * @param message :    消息内容（Json格式）
     * @return Boolean 发送结果
     */
    @Override
    @Async("taskExecutor")
    public void sendMsg(String routingKey, Object message) {
        /* 消息唯一ID */
        String msgId = UUID.randomUUID().toString();
        CorrelationData correlationData = new CorrelationData(msgId);
        String messageText = JSONObject.toJSONString(message);
        // 消息发送，带MessageId
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message msg) throws AmqpException {
                final MessageProperties messageProperties = msg.getMessageProperties();
                // 消息id
                messageProperties.setMessageId(msgId);
                return msg;
            }
        };
        /* 生成Mongo记录 */
        MgMessageQueueLog mongoEntity = new MgMessageQueueLog();
        mongoEntity.setId(correlationData.getId());
        mongoEntity.setExchange("");
        mongoEntity.setRoutingKey(routingKey);
        mongoEntity.setDispatchSuccess(false);
        mongoEntity.setMessageBody(messageText);
        mongoEntity.setCreateTime(new Date());
        mongoTemplate.insert(mongoEntity);
        /* 投递消息 */
        rabbitTemplate.convertAndSend(routingKey, (Object) messageText,messagePostProcessor, correlationData);
    }

    @Override
    @Async("taskExecutor")
    public void resendMsg(String routingKey, Object message, CorrelationData correlationData) {
        String messageText = JSONObject.toJSONString(message);
        // 消息发送，带MessageId
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message msg) throws AmqpException {
                final MessageProperties messageProperties = msg.getMessageProperties();
                // 消息id
                messageProperties.setMessageId(correlationData.getId());
                return msg;
            }
        };
        /* 投递消息 */
        rabbitTemplate.convertAndSend(routingKey, (Object) messageText,messagePostProcessor, correlationData);
    }

    /**
     * RabbitMQ发送消息(指定交换机)
     *
     * @param exchange:   交换机
     * @param routingKey: 路由键
     * @param message:    消息内容
     */
    @Override
    @Async("taskExecutor")
    public void sendMsg(String exchange, String routingKey, String message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }

    /**
     * 功能描述:
     * 〈RabbitMQ发送延迟消息〉
     *
     * @param exchange:   交换机
     * @param routingKey: 路由键
     * @param message     消息内容
     * @param delay       延迟时间
     */
    @Override
    @Async("taskExecutor")
    public void sendMsg(String exchange, String routingKey, String message, Long delay) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message, message1 -> {
            message1.getMessageProperties().setHeader("x-delay", delay);
            return message1;
        });
    }

    /**
     * 创建交换机
     *
     * @param exchangeName 交换机名称
     * @param durable      是否长期有效
     */
    @Override
    @Async("taskExecutor")
    public void addExchange(String exchangeName,
                            Boolean durable) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        TopicExchange topicExchange = new TopicExchange(exchangeName, durable, RabbitConfigConstant.AUTO_DELETE);
        rabbitAdmin.declareExchange(topicExchange);
    }

    /**
     * 创建队列
     *
     * @param queueName 队列名称
     * @param durable   声明一个持久队列
     * @param exclusive 声明一个独立队列
     */
    @Override
    @Async("taskExecutor")
    public void addQueue(String queueName,
                         Boolean durable,
                         Boolean exclusive) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        Queue queue = new Queue(queueName, durable, exclusive, RabbitConfigConstant.AUTO_DELETE);
        rabbitAdmin.declareQueue(queue);
    }

    /**
     * 消息队列、交换机、路由绑定
     *
     * @param exchangeName 交换机名称
     * @param routingKey:  路由键
     * @param queueName    队列名称
     * @param durable      声明一个持久队列/是否长期有效
     */
    @Override
    @Async("taskExecutor")
    public void declareTopicRoutingQueue(String exchangeName,
                                         String routingKey,
                                         String queueName,
                                         Boolean durable) {
        //对交换机名称校验 为null则设置为默认
        if (StringUtils.isBlank(exchangeName)) {
            exchangeName = RabbitConfigConstant.DEFAULT_EXCHANGE_NAME;
        }

        //对路由键校验 为null则设置为默认
        if (StringUtils.isBlank(routingKey)) {
            routingKey = RabbitConfigConstant.DEFAULT_ROUTING_KEY;
        }

        //对队列名称校验 为null则设置为默认
        if (StringUtils.isBlank(queueName)) {
            routingKey = RabbitConfigConstant.DEFAULT_QUEUE;
        }

        //对持久化校验 为null则设置为默认
        if (durable == null) {
            durable = RabbitConfigConstant.DURABLE;
        }
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        TopicExchange topicExchange = new TopicExchange(exchangeName, durable, RabbitConfigConstant.AUTO_DELETE);
        rabbitAdmin.declareExchange(topicExchange);
        Queue queue = new Queue(queueName, durable, RabbitConfigConstant.EXCLUSIVE, RabbitConfigConstant.AUTO_DELETE);
        rabbitAdmin.declareQueue(queue);
        rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(topicExchange).with(routingKey));
    }

    /**
     * 根据队列名称查询队列的数量
     *
     * @param queueName 队列名称
     */
    @Override
    @Async("taskExecutor")
    public Integer findMessageCount(String queueName) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        return rabbitAdmin.getRabbitTemplate().execute(channel -> {
            final AMQP.Queue.DeclareOk ok = channel.queueDeclare(queueName, true, false, false, null);
            return ok.getMessageCount();
        });
    }

    /**
     * 删除消息队列
     *
     * @param queueName 队列名称
     */
    @Override
    @Async("taskExecutor")
    public void deleteQuery(String queueName) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.deleteQueue(queueName);
    }

    /**
     * 删除交换机
     *
     * @param exchangeName 交换机名称
     */
    @Override
    @Async("taskExecutor")
    public void deleteExchangeName(String exchangeName) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.deleteExchange(exchangeName);
    }
}
